package com.aotain.nyx.xss;

import java.text.SimpleDateFormat;
import java.util.ArrayList;
import java.util.Date;
import java.util.List;

import org.apache.commons.codec.binary.Base64;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.util.Collector;
import org.apache.log4j.Logger;

import com.aotain.common.mongo.DataColumn;
import com.aotain.apollo.mongodb.MongoImportTool;
import com.aotain.nyx.statis.AbnStatisTuple;

/**
 * 
     *@Title:跨站脚本攻击检测方法
     *@Description:
     *@Author:Turk
     *@Since:2017年7月12日
     *@Version:1.1.0
 */
public class XSSDetect implements FlatMapFunction<Tuple2<String,XSSTuple>, AbnStatisTuple>{

	/**
	 * 一分钟内发现相同SQL注入行为的次数
	 */
	private int threshold = 3;
	
	/**
	 * 
	 */
	private static final long serialVersionUID = -5427734737259338211L;

	@Override
	public void flatMap(Tuple2<String, XSSTuple> value, Collector<AbnStatisTuple> out)
			throws Exception {
		// TODO Auto-generated method stub
		XSSTuple tuple = value.f1;
		int urlPVNum = tuple.getCounter();
				
		if(urlPVNum >= threshold){
			System.out.println("XSSDetect target >>>>>>>>> "+value.f0 + " >>>>>>>>> matchNum:" 
					+  urlPVNum);
					
			String dip = tuple.getDestIP();
			String dport = tuple.getDestPort();
			String sip = tuple.getSourceIP();
			String sport = tuple.getSourcePort();
			String gis = new String(Base64.decodeBase64(tuple.getGIS()));
			String type = tuple.getType();
			String idc = tuple.getIDC();
			String[] gisArray = gis.split("#");
			String destAreaName = "";
			String destGis = "";
			String sourceAreaName = "";
			String sourceGis = "";
			String sourceAreaCountry = "";
			String sourceAreaId = "";
			String sourceProvinceName = "";
			if(gisArray.length > 1) {
				destAreaName = gisArray[0];
				destGis = gisArray[1];
				sourceAreaName = gisArray[2];
				sourceGis = gisArray[3];
				sourceAreaCountry = gisArray[4];
				sourceAreaId = gisArray[5];
				sourceProvinceName =  gisArray[6].trim(); //如果省为空，精确到国家
			} else {
				Logger.getLogger(XSSDetect.class).warn("GIS Object is null: " + gis);
			}
				
//						String sourceAreaCityId = gisArray[7];
//						String sourceAreaProvinceId = gisArray[8];
				
			String sAbnormalTable = "SDS_ABNORMAL_LOG2";
						
			String accesstime = new SimpleDateFormat("yyyyMMddHHmmss").format(new Date(System.currentTimeMillis()+15000));
						//源IP发起的攻击行为
			String rowKeyAbnormal = dip + "_" + accesstime + "_" + sip + "_"  +  dport + "_10";
					
			MongoImportTool importtool = MongoImportTool.getInstance();
			List<DataColumn> row = new ArrayList<DataColumn>();
			row.add(new DataColumn("ROWKEY", rowKeyAbnormal)); //主键字段，用于更新记录
			row.add(new DataColumn("SOURCEIP", sip));
			row.add(new DataColumn("DESTPORT", dport));
			row.add(new DataColumn("ACCESSTIME", Long.parseLong(accesstime)));
			row.add(new DataColumn("ABRNORMAL", 10));
			row.add(new DataColumn("DESTIP", dip));
			row.add(new DataColumn("SOURCEAREA", sourceAreaName));
			row.add(new DataColumn("SOURCEGEO", sourceGis));
			row.add(new DataColumn("SOURCECOUNTRY", sourceAreaCountry));
			row.add(new DataColumn("DESTAREA", destAreaName));
			row.add(new DataColumn("DESTGEO", destGis));
			row.add(new DataColumn("ATTNUM", urlPVNum));
			row.add(new DataColumn("DESC", "正在进行  XSS跨站脚本攻击  " + tuple.getURL()));
			row.add(new DataColumn("EVALUATE", "40"));
			row.add(new DataColumn("ENDTIME", accesstime));
			if(sourceProvinceName.equals("")) {
				sourceProvinceName = sourceAreaCountry;
			}
			row.add(new DataColumn("PROVINCE", sourceProvinceName));
			row.add(new DataColumn("UPSTREAMOCTETS", tuple.getUpStreamOctets()));
			row.add(new DataColumn("UPSTREAMPACKET", tuple.getUpStreamPacket()));
			row.add(new DataColumn("ATTTYPE", "2"));
			row.add(new DataColumn("FLOWDIRECTION", tuple.getFlowDirection()));
			
			importtool.InsertRowData(sAbnormalTable, row);	 
			
			AbnStatisTuple abnTuple = new AbnStatisTuple();
			abnTuple.setDestIP(dip);
			abnTuple.setSourceIP(sip);
			abnTuple.setType("XSS");
			abnTuple.setUpStreamOctets(tuple.getUpStreamOctets());
			abnTuple.setUpStreamPacket(tuple.getUpStreamPacket());
			abnTuple.setAttackNum(1l);
			abnTuple.setSourceCountry(sourceAreaCountry);
			abnTuple.setSourceProvince(sourceProvinceName);
			abnTuple.setAttackType("2");
			abnTuple.setAttTime(accesstime);
			out.collect(abnTuple);
		}		
		
	}

}
